package ranks.tasks.calc;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextOutputFormat;

import ranks.base.Data;
import ranks.tasks.readable.Readable;

public class Calc extends Job {

    private int round;

    private String output;

    private String[] settings;

    public Calc(int round, String... settings) {
        super("Rank Calc Round " + round, settings);

        this.round = round;
        this.settings = settings;
        this.output = this.getConfig("-output");
    }

    @Override
    protected CombineConfig getCombineConfig() {
        return CombineConfig.newInstance(Calc_Combiner.class);
    }

    @Override
    protected InputConfig getInputConfig(FileSystem fs) throws IOException {
        return InputConfig.newInstance(KeyValueTextInputFormat.class, output
                + "/Round_" + (round - 1));
    }

    @Override
    protected MapConfig getMapConfig() {
        return MapConfig.newInstance(Calc_Mapper.class, 0, Text.class,
                Text.class, DefaultCodec.class);
    }

    @Override
    protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
        return OutputConfig.newInstance(TextOutputFormat.class, Text.class,
                Text.class, output + "/Round_" + round, GzipCodec.class);
    }

    @Override
    protected ReduceConfig getReduceConfig() {
        return ReduceConfig.newInstance(Calc_Reducer.class, 1000);
    }

    @Override
    protected void setUp(JobConf conf, FileSystem fs) throws IOException {
        this.setJobConfConfig("Output", output);
        this.setJobConfConfig("Round", round);

        long websize = Long.parseLong(new BufferedReader(new InputStreamReader(
                fs.open(new Path(output + "/WebSize")))).readLine());
        this.setJobConfConfig("WebSize", websize);

        if (round == 1) {
            this.setJobConfConfig("Z", new Data().toString());
        } else {
            String z = "";
            try {
                z = new BufferedReader(new InputStreamReader(fs.open(new Path(
                        output + "/Z_" + (round - 1))))).readLine();
            } catch (IOException e) {
                z = new Data().toString();
            }
            this.setJobConfConfig("Z", z);
        }
    }

    @Override
    protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
            throws IOException {
        long delta = this.getCounterSumByGroup(job, "Delta");
        long distance = this.getCounterSumByGroup(job, "Distance");

        if (distance <= delta) {
            try {
                new Readable(round, settings).submit();
            } catch (Exception e) {
                throw new IOException(e);
            }
        }

        fs.delete(new Path(output + "/Round_" + (round - 1)), true);
        fs.delete(new Path(output + "/Z_" + (round - 1)), true);
    }

}
